package com.atguigu.flink.chapter05.sink;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/15 11:06
 */
public class Flink02_Sink_Redis_1 {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
    
        DataStreamSource<WaterSensor> s1 = env.fromElements(
            new WaterSensor("sensor_1", 1L, 10),
            new WaterSensor("sensor_1", 2L, 20),
            new WaterSensor("sensor_1", 5L, 20),
            new WaterSensor("sensor_2", 3L, 30),
            new WaterSensor("sensor_1", 4L, 40)
        );
    
        FlinkJedisConfigBase redisConf = new FlinkJedisPoolConfig.Builder()
            .setHost("hadoop162")
            .setPort(6379)
            .setMaxTotal(100)
            .setMaxIdle(10)
            .setDatabase(0)
            .setTimeout(10 * 1000)
            .setMinIdle(1)
            .build();
        
        s1
            .keyBy(WaterSensor::getId)
            .sum("vc")
            .addSink(new RedisSink<>(redisConf, new RedisMapper<WaterSensor>() {
                // 返回一个:命令描述符
                @Override
                public RedisCommandDescription getCommandDescription() {
                    // 参数2: 只有当是hash和zset的时候有效
                    return new RedisCommandDescription(RedisCommand.SET, "");
                }
            
                // 从数据中获取一个key
                @Override
                public String getKeyFromData(WaterSensor data) {
                    return data.getId();
                }
            
                // 获取最终的那个写入具体的数据
                @Override
                public String getValueFromData(WaterSensor data) {
                    return data.getVc() + "";
                }
                
            }));
        
    
        env.execute();
    }
}
/*
string

key         value

sensor_1    100

list

set hash zset


 */
